package com.tbit.uqbike.config.rabbitmq;

import cn.hutool.core.util.StrUtil;
import com.tbit.common.constant.CommonConstant;
import com.tbit.common.utils.TraceUtil;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * @author chenzhuang
 * @Description:
 * @date 2022/7/28 10:23
 */
@Configuration
public class RabbitConfig {
    @Value("${mq.exchange}")
    private String MQ_EXCHANGE;
    @Value("${mq.rideexchange}")
    private String MQ_RIDEEXCHANGE;
    @Value("${mq.queue}")
    private String MQ_QUEUE;
    @Value("${mq.ddqueue}")
    private String MQ_DDQUEUE;
    @Value("${mq.aiqueue}")
    private String MQ_AIQUEUE;
    @Value("${mq.ridequeue}")
    private String MQ_RIDEQUEUE;
    @Value("${mq.modelqueue}")
    private String MQ_MODELQUEUE;
    @Value("${mq.payqueue}")
    private String MQ_PAYQUEUE;
    @Value("${mq.eventqueue}")
    private String MQ_EVENTQUEUE;
    @Value("${mq.payexchange}")
    private String MQ_PAYEXCHANGE;
    @Value("${mq.eventkey}")
    private String eventkey;
    @Value("${mq.orderqueue}")
    private String MQ_ORDERQUEUE;
    @Value("${mq.orderkey}")
    private String MQ_ORDERKEY;
    @Value("${mq.orderexchange}")
    private String MQ_ORDEREXCHANGE;
    @Value("${mq.ttl}")
    private Long MQ_TTL;


    @Value("${mq.logexchange}")
    private String MQ_LOGEXCHANGE;
    @Value("${mq.logkey}")
    private String LOGKEY;
    @Value("${mq.logqueue}")
    private String MQ_LOGQUEUE;

    @Value("${mq.slskey}")
    private String SLSKEY;
    @Value("${mq.slsqueue}")
    private String MQ_SLSQUEUE;
    @Value("${mq.cashkey}")
    private String CASHKEY;
    @Value("${mq.cashqueue}")
    private String MQ_CASHQUEUE;
    @Value("${mq.returndepositkey}")
    private String RETURNDEPOSITKEY;
    @Value("${mq.returndepositqueue}")
    private String MQ_RETURNDEPOSITQUEUE;
    @Value("${mq.exceptionKey}")
    private String EXCEPTIONKEY;
    @Value("${mq.exceptionQueue}")
    private String EXCEPTIONQUEUE;

    @Bean
    public CachingConnectionFactory connectionFactory(RabbitProperties rabbitProperties) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitProperties.getHost(), rabbitProperties.getPort());
        //CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        //connectionFactory.setAddresses(rabbitProperties.getAddresses());
        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.setVirtualHost(StringUtils.hasText(rabbitProperties.getVirtualHost()) ? rabbitProperties.getVirtualHost() : "/");
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin amqpAdmin(CachingConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.declareExchange(exchange());
        rabbitAdmin.declareExchange(rideExchange());
        rabbitAdmin.declareExchange(payexchange());
        rabbitAdmin.declareExchange(orderExchange());
        rabbitAdmin.declareExchange(logExchange());
        rabbitAdmin.declareQueue(mqQueue());
        rabbitAdmin.declareQueue(ddQueue());
        rabbitAdmin.declareQueue(aiQueue());
        rabbitAdmin.declareQueue(modelQueue());
        rabbitAdmin.declareQueue(eventQueue());
        rabbitAdmin.declareQueue(payQueue());
        rabbitAdmin.declareQueue(orderQueue());
        rabbitAdmin.declareQueue(logQueue());
        rabbitAdmin.declareQueue(slsQueue());
        rabbitAdmin.declareQueue(cashQueue());
        rabbitAdmin.declareQueue(depositQueue());
        rabbitAdmin.declareBinding(bindExchange1());
        rabbitAdmin.declareBinding(bindExchange2());
        rabbitAdmin.declareBinding(bindExchange3());
        rabbitAdmin.declareBinding(bindExchange5());
        rabbitAdmin.declareBinding(bindExchange6());
        rabbitAdmin.declareBinding(bindExchange7());
        rabbitAdmin.declareBinding(bindExchange8());
        rabbitAdmin.declareBinding(bindExchange9());
        rabbitAdmin.declareBinding(bindCashExchange());
        rabbitAdmin.declareBinding(bindDepositExchange());
        rabbitAdmin.declareBinding(bindExchange10());
        rabbitAdmin.declareQueue(exceptionQueue());
        rabbitAdmin.declareBinding(bindExceptionExchange());
        return rabbitAdmin;
    }
    @Bean(name = {"rabbitListenerContainerFactory"})
    @ConditionalOnProperty(
            prefix = "spring.rabbitmq.listener",
            name = {"type"},
            havingValue = "simple",
            matchIfMissing = true
    )
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, CachingConnectionFactory connectionFactory, ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory,connectionFactory);
        simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);

        // 使用RabbitTemplate接收消息前会调用这个方法
        factory.setAfterReceivePostProcessors(message -> {
            Object header = message.getMessageProperties().getHeader(CommonConstant.TRACE_ID);
            String traceId = String.valueOf(header);
            if (Objects.isNull(header) || StrUtil.isBlank(traceId)){
                traceId = TraceUtil.buildTraceId();
            }
            TraceUtil.setTraceId(traceId);
            return message;
        });
        return factory;
    }

    @Bean
    @Primary
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        rabbitTemplate.setBeforePublishPostProcessors(new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                return message;
            }

            // 使用RabbitTemplate发送消息前会调用这个方法
            @Override
            public Message postProcessMessage(Message message, Correlation correlation, String exchange, String routingKey) {
                String traceId = TraceUtil.getTraceIdByLocal();
                if (StrUtil.isBlank(traceId)){
                    traceId = TraceUtil.buildTraceId();
                }
                message.getMessageProperties().setHeader(CommonConstant.TRACE_ID,traceId);
                return message;
            }
        });

        rabbitTemplate.setExchange(MQ_EXCHANGE);
        return rabbitTemplate;
    }


    @Bean
    public Queue mqQueue() {
        return new Queue(MQ_QUEUE, true, false, false);
    }

    @Bean
    public Queue ddQueue() {
        return new Queue(MQ_DDQUEUE, true, false, false);
    }

    @Bean
    public Queue aiQueue() {
        return new Queue(MQ_AIQUEUE, false, false, false);
    }

    @Bean
    public Queue modelQueue() {
        return new Queue(MQ_MODELQUEUE, false, false, false);
    }

    @Bean
    public Queue eventQueue() {
        return new Queue(MQ_EVENTQUEUE, false, false, false);
    }

    @Bean
    public Queue payQueue() {
        return new Queue(MQ_PAYQUEUE, true, false, false);
    }

    @Bean
    public Queue orderQueue() {
        return new Queue(MQ_ORDERQUEUE, true, false, false);
    }

    @Bean
    public Queue logQueue() {
        return new Queue(MQ_LOGQUEUE, true, false, false);
    }

    @Bean
    public Queue cashQueue() {
        return new Queue(MQ_CASHQUEUE, true, false, false);
    }

    @Bean
    public Queue slsQueue() {
        return new Queue(MQ_SLSQUEUE, false, false, false);
    }
    @Bean
    public Queue depositQueue() {
        return new Queue(MQ_RETURNDEPOSITQUEUE, false, false, false);
    }
    @Bean
    public Queue exceptionQueue() {
        return new Queue(EXCEPTIONQUEUE, false, false, false);
    }

    @Bean
    public Exchange exchange() {
        /** direct交换机 不显式指定路由键则默认为队列名 */
        return new DirectExchange(MQ_EXCHANGE, false, false);
    }

    @Bean
    public Exchange payexchange() {
        /** direct交换机 不显式指定路由键则默认为队列名 */
        return new DirectExchange(MQ_PAYEXCHANGE, true, false);
    }

    @Bean
    public Exchange orderExchange() {
        /** direct交换机 不显式指定路由键则默认为队列名 */
        return new DirectExchange(MQ_ORDEREXCHANGE, true, false);
    }

    @Bean
    public Exchange logExchange() {
        /** direct交换机 不显式指定路由键则默认为队列名 */
        return new DirectExchange(MQ_LOGEXCHANGE, true, false);
    }

    @Bean
    public FanoutExchange rideExchange() {
        return new FanoutExchange(MQ_RIDEEXCHANGE, false, false);
    }

    @Bean
    public Binding bindExchange1() {
        return BindingBuilder.bind(mqQueue()).to(exchange()).with(MQ_QUEUE).noargs();
    }

    @Bean
    public Binding bindExchange2() {
        return BindingBuilder.bind(ddQueue()).to(exchange()).with(MQ_DDQUEUE).noargs();
    }

    @Bean
    public Binding bindExchange3() {
        return BindingBuilder.bind(aiQueue()).to(exchange()).with(MQ_AIQUEUE).noargs();
    }

    @Bean
    public Binding bindExchange5() {
        return BindingBuilder.bind(modelQueue()).to(rideExchange());
    }

    @Bean
    public Binding bindExchange6() {
        return BindingBuilder.bind(payQueue()).to(payexchange()).with(MQ_PAYQUEUE).noargs();
    }

    @Bean
    public Binding bindExchange7() {
        return BindingBuilder.bind(eventQueue()).to(exchange()).with(eventkey).noargs();
    }

    @Bean
    public Binding bindExchange8() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(MQ_ORDERKEY).noargs();
    }

    @Bean
    public Binding bindExchange9() {
        return BindingBuilder.bind(logQueue()).to(logExchange()).with(LOGKEY).noargs();
    }

    @Bean
    public Binding bindExchange10() {
        return BindingBuilder.bind(slsQueue()).to(exchange()).with(SLSKEY).noargs();
    }

    @Bean
    public Binding bindDepositExchange() {
        return BindingBuilder.bind(depositQueue()).to(payexchange()).with(RETURNDEPOSITKEY).noargs();
    }
    @Bean
    public Binding bindCashExchange() {
        return BindingBuilder.bind(cashQueue()).to(payexchange()).with(CASHKEY).noargs();
    }
    @Bean
    public Binding bindExceptionExchange() {
        return BindingBuilder.bind(exceptionQueue()).to(exchange()).with(EXCEPTIONKEY).noargs();
    }
}
